#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <string.h>
#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <signal.h>
#include <errno.h>
#include <assert.h>

#include "base.h"
#include "service.h"
#include "list.h"

static SERVICE service;	//BSS will be zero initialized
static SERVICE * srv = &service;

static int service_send_log(unsigned int level, const char * fmt, ...)
{
	assert(srv);

	if (level > srv->level) return 1;

	if (!srv->loggings[level] && 
		!(srv->loggings[level] = logging_new("stderr", level, LOG_BUFFER_SIZE))) {
		fprintf(stderr, "#service_send_log# logging_new failed\n");
		return -1;
	}

	LOGGING * logging = srv->loggings[level];

	va_list argp;
	va_start(argp, fmt);

	/* send log to thread logger */
	TASK_LOG * data = malloc(sizeof(struct TASK_LOG));
	if (!data) {
		fprintf(stderr, "#service_send_log# alloc data failed\n");
		return -2;
	}

	data->level = level;
	data->buf = malloc(logging->bufsz);
	if (!data->buf) {
		fprintf(stderr, "#service_send_log# alloc msg failed\n");
		return -3;
	}

	int len = snprintf(data->buf, logging->bufsz - 1, "{%lx}", pthread_self());
	len += vsnprintf(data->buf + len, logging->bufsz - 1 - len, fmt, argp);
	data->buf[len > logging->bufsz - 1 ? logging->bufsz - 1 : len] = '\0';

	va_end(argp);

	/* thread logger haven't start up, write log directly */
	if (!srv->logger || !srv->logger->running) {
		rlogit(logging, data->level, data->buf);
		free(data->buf);
		free(data);
		return 2;
	}


	TASK * task = malloc(sizeof(struct TASK));
	if (!task) {
		fprintf(stderr, "#service_send_log# alloc task failed\n");
		return -3;
	}
	task->data = data;

	if (send_request(task, srv->logger) < 0) {
		fprintf(stderr, "#service_send_log# send task failed\n");
		return -4;
	}

	return 0;
}

//called by thread_notify_handle
static int logger_ontask(TASK * task, const THREAD * logger)
{
	assert(task && task->data);
	
	TASK_LOG * data = task->data;

	LOGGING * logging = srv->loggings[data->level];

	assert(logging);
		
	rlogit(logging, data->level, data->buf);

	free(data->buf);
	free(data);
	free(task);
	return 0;
}

static int logger_init(const THREAD * logger)
{
	return 0;
}

static int logger_clear(const THREAD * logger)
{
	return 0;
}

//bind with thread response_event
static void master_notify_handler(int fd, short which, void *arg)
{
	assert(fd && arg);
	THREAD * th = (THREAD *)arg;

	TASK * task = recv_response(th);
	if (!task) {
		return;
	}
	if (srv->master->ontask(task, srv->master) < 0) {
		return;
	}
}

static int pid_lock()
{
	if (srv->pidpath[0] == '\0')
		strcpy(srv->pidpath, SRV_PID_PATH);

	srv->pid = getpid();

	if((srv->pidfd = open(srv->pidpath, O_CREAT | O_RDWR, 0644)) == -1) {
		send_log(LV_SYS, "open %s: %s", srv->pidpath, strerror(errno));
		return -1;
	}

	char spid[32];
	int len;

	struct flock lock = {F_WRLCK, SEEK_SET, 0, 0};
	if (fcntl(srv->pidfd, F_SETLK, &lock) == -1) {
		send_log(LV_SYS, "lock %s: %s", srv->pidpath, strerror(errno));

		if ((len = read(srv->pidfd, spid, 32)) > 0) {
			spid[--len] = '\0';
			send_log(LV_SYS, "service pid %s maybe exist", spid);
		}

		close(srv->pidfd);
		return -2;
	}

	len = sprintf(spid, "%d\n", srv->pid);
	if (write(srv->pidfd, spid, len) != len) {
		send_log(LV_SYS, "write %s: %s", srv->pidpath, strerror(errno));
		close(srv->pidfd);
		return -3;
	}
	return 0;
}

static int pid_unlock()
{
	if (!srv->pidfd || !srv->pidpath[0]) return -1;

	struct flock lock = {F_UNLCK, SEEK_SET, 0, 0};
	if (fcntl(srv->pidfd, F_SETLK, &lock) < 0)
		send_log(LV_SYS, "unlock %s: %s\n", srv->pidpath, strerror(errno));

	close(srv->pidfd);
	srv->pidfd = 0;

	unlink(srv->pidpath);
	memset(srv->pidpath, 0, SRV_PATH_SIZE);
	return 0;
}

static void service_stop()
{
	if (srv->stat != SRV_RUNNING && srv->stat != SRV_SHUTTING) exit(20);

	event_base_loopexit(srv->master->base, NULL);
	srv->stat = SRV_SHUTTING;
}

static void signal_handler(int signo)
{
	switch (signo) {
	case SIGINT:
		send_log(LV_SYS, "Signal [SIGINT] received, System exit...");
		service_stop();
		break;
	case SIGHUP:
		send_log(LV_SYS, "Signal [SIGHUP] received, System exit...");
		service_stop();
		break;
	case SIGPIPE:
		send_log(LV_SYS, "Signal [SIGPIPE] received, Ignored...");
		break;
	default:
		send_log(LV_SYS, "Signal [%d] received, Ignored...", signo);
		break;
	}
	return;
}

typedef __sighandler_t sighandler_t;
static void sig_set(int signo, sighandler_t handler)
{
	struct sigaction sa = {.sa_flags = 0};
	sa.sa_handler = handler;

	if (sigemptyset(&sa.sa_mask) == -1 || sigaction(signo, &sa, 0) == -1) {
        send_log(LV_INFO, "Signal [%d] set %p", signo, handler);
		exit(1);
    }
}

static int timing_clear()
{
	if (!srv->timings_num) return 1;
	int i;
	for (i = 0; i < srv->timings_num; i ++) {
		if(srv->timings[i]) timing_free(srv->timings[i]);
	}

	return 0;
}

static int timing_init()
{
	if (!srv->timings_num) return 1;

	if (!srv->timer) {
		send_log(LV_SYS, "create thread timer first");
		return -1;
	}

	int i;
	TIMING_ARG * arg;
	for (i = 0; i < srv->timings_num; i ++) {
		arg = srv->timings_arg + i;
		srv->timings[i] = timing_new(arg->usec, arg->ontiming, srv->timer->base, srv->timer);
		if (!srv->timings[i]) {
			send_log(LV_SYS, "#timer_init# timing_new()");
			return -2;
		}
	}
	srv->timer->sys = srv->timings;
	return 0;
}

static int threads_clear()
{
	if (srv->timer) thread_shutdown(srv->timer);

	threads_wait(srv->workers_num + 1);		//make sure timer is stop

	int i;
	for (i = 0; i < srv->workers_num; i ++) {
		thread_shutdown(srv->workers[i]);
	}

	threads_wait(1);	//make sure workers are stop

	if (srv->logger) thread_shutdown(srv->logger);

	if (srv->master->clear && srv->master->clear(srv->master) < 0) {
		send_log(LV_ERROR, "%s clear", TTAG(srv->master));
	}

	threads_wait(0);	//make sure logger is stop

	timing_clear(); 
	thread_free(srv->timer); srv->timer = NULL;
	thread_free(srv->logger); srv->logger = NULL;
	thread_free(srv->master); srv->master = NULL;

	for (i = 0; i < srv->workers_num && srv->workers[i]; i ++) {
		event_del(&srv->workers[i]->response_event);
		thread_free(srv->workers[i]);
		srv->workers[i] = NULL;
	}
	srv->workers_num = 0;

	return 0;
}

static int threads_init()
{
	assert(srv->workers_num < WORKERS_NUM_MAX);

	//new logger
	assert(!srv->logger);
	srv->logger = thread_new(logger_ontask, logger_init, logger_clear, srv->loggings);
	if (!srv->logger) {
		send_log(LV_SYS, "logger new failed");
		threads_clear();
		return -1;
	}

	THREAD_ARG * arg = &srv->master_arg;

	//new master
	assert(!srv->master);
	if ((arg->ontask || arg->init || arg->clear) && 
			!(srv->master=thread_new(arg->ontask, arg->init, arg->clear, srv))) {
		send_log(LV_SYS, "master new failed");
		threads_clear();
		return -2;
	}

	if (srv->master && srv->master->init && srv->master->init(srv->master) < 0) {
		send_log(LV_ERROR, "%s init", TTAG(srv->master));
		threads_clear();
		return -3;
	}

	//news workers
	srv->next_workeri = 0;

	int i;
	for (i = 0; i < srv->workers_num; i ++) {
		arg = srv->workers_arg + i;
		assert(arg->ontask && !srv->workers[i]);

		srv->workers[i] = thread_new(arg->ontask, arg->init, arg->clear, NULL);
		if (!srv->workers[i]) {
			send_log(LV_SYS, "workers new failed");
			threads_clear();
			return -4;
		}
		
		event_set(&srv->workers[i]->response_event, 
					srv->workers[i]->response_recv_fd,
					EV_READ | EV_PERSIST, master_notify_handler, srv->workers[i]);

		event_base_set(srv->master->base, &srv->workers[i]->response_event);

		if (event_add(&srv->workers[i]->response_event, 0) == -1) {
			send_log(LV_SYS, "monitor libevent notify pipe on worker failed");
			threads_clear();
			return -5;
		}
	}

	//new timer
	assert(!srv->timer);
	if (srv->timings_num) {
		srv->timer = thread_new(NULL, srv->timer_arg.init, srv->timer_arg.clear, NULL);
		if (!srv->timer) {
			send_log(LV_SYS, "timer new failed");
			threads_clear();
			return -6;
		}

		event_set(&srv->timer->response_event, 
					srv->timer->response_recv_fd,
					EV_READ | EV_PERSIST, master_notify_handler, srv->timer);

		event_base_set(srv->master->base, &srv->timer->response_event);

		if (event_add(&srv->timer->response_event, 0) == -1) {
			send_log(LV_SYS, "monitor libevent notify pipe on timer failed");
			threads_clear();
			return -7;
		}
	}

	if (timing_init() < 0) {
		send_log(LV_SYS, "timings init failed");
		threads_clear();
		return -8;
	}

	/* threads start up */
	if (thread_startup(srv->logger) < 0) {
		send_log(LV_SYS, "logger start up failed");;
		threads_clear();
		return -9;
	}
	threads_wait(1);	//make sure logger is running

	for (i=0; i<srv->workers_num; i++) {
		if (thread_startup(srv->workers[i]) < 0) {
			send_log(LV_SYS, "worker %d start up failed", i);
			threads_clear();
			return -10;
		}
	}
	threads_wait(srv->workers_num + 1); //make sure workers are running

	if (srv->timings_num && thread_startup(srv->timer) < 0) {
		send_log(LV_SYS, "timer start up failed");;
		threads_clear();
		return -11;
	}
	return 0;
}

static int ifaces_clear()
{
	int i;
	for (i = 0; i < srv->ifaces_num && srv->ifaces[i]; i ++) {
		conn_close(srv->ifaces[i]);
		srv->ifaces[i] = NULL;
	}
	return 0;
}

static int ifaces_init()
{
	assert(srv->ifaces_num < IFACES_NUM_MAX);

	IFACE_ARG * arg;
	int sfd, i, init_state, event_flags;
	struct sockaddr_in addr;
	for (i = 0; i < srv->ifaces_num; i ++) {
		arg = srv->ifaces_arg + i;

		assert(arg->addr[0] && arg->port && arg->onpack && arg->depack && !srv->ifaces[i]);


		switch (arg->type) {
		case TYPE_SERVER:
			sfd = server_socket(arg->addr, arg->port, &addr);
			init_state = conn_listening;
			event_flags = EV_READ | EV_PERSIST;
			break;
		case TYPE_CLIENT:
			sfd = client_socket(arg->addr, arg->port, &addr);
			init_state = conn_connecting;
			event_flags = EV_WRITE | EV_PERSIST;
			break;
		default:
			sfd = 0;
			break;
		}

		if (sfd <= 0) {
			send_log(LV_SYS, "failed to new socket on %s:%d", arg->addr, arg->port);
			ifaces_clear();
			return -2;
		}

		srv->ifaces[i] = conn_new(sfd, &addr, init_state, event_flags, 
				srv->master->base, arg->depack, arg->onpack, srv);
		if (srv->ifaces[i] == NULL) {
			send_log(LV_SYS, "failed to create conn on %s:%d", arg->addr, arg->port);
			ifaces_clear();
			return -3;
		}
	}
	return 0;
}

void service_run()
{
	//init log first, now log without logger thread
	send_log = service_send_log;

	int ret = 0;

	switch (srv->stat) {
	case SRV_READY:
		srv->stat = SRV_RUNNING;
		break;
	case SRV_RUNNING:
		send_log(LV_SYS, "service already exist");
		exit(10);
	default:
		send_log(LV_SYS, "stat %d before service run", srv->stat);
		exit(11);
	}

	if (pid_lock() < 0) exit(12);

	send_log(LV_SYS, "service %d at %s start up", srv->pid, srv->pidpath);


	sig_set(SIGTERM, signal_handler);
	sig_set(SIGTSTP, signal_handler);/* Ctrl+Z */
	sig_set(SIGQUIT, signal_handler);/* Ctrl+\ */
	sig_set(SIGINT,  signal_handler);/* Ctrl+C */
	sig_set(SIGPIPE, signal_handler);
	sig_set(SIGALRM, signal_handler);
	sig_set(SIGHUP,  signal_handler);
	sig_set(SIGCHLD, signal_handler);
	sig_set(SIGUSR2, signal_handler);

	if (threads_init() < 0) {ret = 13; goto STOP;}

	if (ifaces_init() < 0) {ret = 14; goto STOP;}

	event_base_loop(srv->master->base, 0);

STOP:
	ifaces_clear();
	threads_clear();

	send_log(LV_SYS, "service %d at %s shut down", srv->pid, srv->pidpath);
	pid_unlock();

	exit(ret);
}

/* dispatch to workeri */
int service_dispatch_workeri(TASK * task, unsigned int workeri)
{
	assert(task && workeri < srv->workers_num);

	THREAD * worker = *(srv->workers + workeri);
	assert(worker);

//	send_log(LV_DEBUG, "%s dispatch : %s", TTAG(srv->master), (char *)task->data);
	if (send_request(task, worker) < 0) {
		return -1;
	}

	srv->next_workeri = workeri + 1;
	return 0;
}

/* rang is [from, to] */
int service_dispatch_rang(TASK * task, unsigned int from_workeri, unsigned int to_workeri)
{
	assert(task && from_workeri <= to_workeri && to_workeri < srv->workers_num);

	int workeri = srv->next_workeri;

	//out of range
	if (workeri > to_workeri || workeri < from_workeri) workeri = from_workeri;

	return service_dispatch_workeri(task, workeri);
}

/* rang is [0, workers_num - 1] */
int service_dispatch(TASK * task)
{
	assert(task);

	return service_dispatch_rang(task, 0, srv->workers_num - 1);
}

void service_set_log_level(unsigned int level)
{
	if (level > LV_DEBUG) {
		fprintf(stderr, "illegal log level: %d", level);
		exit(1);
	}

	srv->level = level;
}

void service_set_log(const char * path, unsigned int level, unsigned int bufsz)
{
	if (!path || level > LV_DEBUG) {
		fprintf(stderr, "illegal param [path: %s][level: %d]\n", path, level);
		exit(1);
	}

	if (srv->loggings[level]) {
		fprintf(stderr, "logging has been set at [level: %d]\n", level);
		exit(1);
	}

	//set means log it
	if (level > srv->level) {
		srv->level = level;
	}

	if (srv->loggings[level])
		logging_free(srv->loggings[level]);

	if (bufsz == 0 || bufsz > 2<<20)
		bufsz = LOG_BUFFER_SIZE;

	if (!(srv->loggings[level] = logging_new(path, level, bufsz))) {
		fprintf(stderr, "logging new failed [path: %s][level: %d]\n", path, level);
		exit(1);
	}
}

void service_set_master(TH_ONTASK_FUNC ontask, TH_INIT_FUNC init, TH_CLEAR_FUNC clear)
{
	if (!ontask) {
		fprintf(stderr, "master need ontask function\n");
		exit(1);
	}

	if (srv->master_arg.ontask || srv->master_arg.init || srv->master_arg.clear) {
		fprintf(stderr, "master has been set\n");
		exit(1);
	}

	srv->master_arg.ontask = ontask;
	srv->master_arg.init = init;
	srv->master_arg.clear = clear;
}

void service_set_timer(TH_ONTASK_FUNC ontask, TH_INIT_FUNC init, TH_CLEAR_FUNC clear)
{
	if (srv->timer_arg.ontask || srv->timer_arg.init || srv->timer_arg.clear) {
		fprintf(stderr, "timer has been set\n");
		exit(1);
	}

	srv->timer_arg.ontask = ontask;
	srv->timer_arg.init = init;
	srv->timer_arg.clear = clear;
}

void service_add_workers(unsigned int workers_num, 
		TH_ONTASK_FUNC ontask, TH_INIT_FUNC init, TH_CLEAR_FUNC clear)
{
	if (!workers_num) {
		fprintf(stderr, "workers num is 0 ?\n");
		exit(1);
	}

	if (srv->workers_num + workers_num > WORKERS_NUM_MAX) {
		fprintf(stderr, "too much workers: %d + %d > %d\n", 
				srv->workers_num, workers_num, WORKERS_NUM_MAX);
		exit(1);
	}
	
	if (!ontask) {
		fprintf(stderr, "worker need ontask function\n");
		exit(1);
	}

	while (workers_num --) {
		srv->workers_arg[srv->workers_num].ontask = ontask;
		srv->workers_arg[srv->workers_num].init = init;
		srv->workers_arg[srv->workers_num].clear = clear;
		srv->workers_num ++;
	}
}

void service_add_iface(enum SOCKET_TYPE type, const char * ip, unsigned int port, 
						DEPACK_FUNC depack, ONPACK_FUNC onpack)
{
	if (srv->ifaces_num + 1 == IFACES_NUM_MAX) {
		fprintf(stderr, "too many ifaces, max: %d\n", IFACES_NUM_MAX);
		exit(1);
	}

	if (!ip || !ip[0] || strlen(ip) > IPADDR_SIZE - 1 || !port) {
		fprintf(stderr, "illegal param [ip: %s][port: %d]\n", ip, port);
		exit(1);
	}

	if (!depack || !onpack) {
		fprintf(stderr, "depack and onpack function is needed\n");
		exit(1);
	}

	srv->ifaces_arg[srv->ifaces_num].type = type;
	strncpy(srv->ifaces_arg[srv->ifaces_num].addr, ip, IPADDR_SIZE - 1);
	srv->ifaces_arg[srv->ifaces_num].port = port;
	srv->ifaces_arg[srv->ifaces_num].depack = depack;
	srv->ifaces_arg[srv->ifaces_num].onpack = onpack;
	
	srv->ifaces_num ++;
}

void service_add_timing(unsigned int usec, ONTIMING_FUNC ontiming, 
						TH_ONTASK_FUNC ontask, TH_INIT_FUNC init, TH_CLEAR_FUNC clear)
{
	if (srv->timings_num + 1 == TIMINGS_NUM_MAX) {
		fprintf(stderr, "too many timings, max: %d\n", TIMINGS_NUM_MAX);
		exit(1);
	}

	if (!usec) {
		fprintf(stderr, "illegal interval micro seconds: %d\n", usec);
		exit(1);
	}

	if (!ontiming) {
		fprintf(stderr, "ontiming function is needed\n");
		exit(1);
	}

	srv->timings_arg[srv->timings_num].usec = usec;
	srv->timings_arg[srv->timings_num].ontiming = ontiming;
	srv->timings_num ++;
}

void service_set_pid_path(const char * path)
{
	if (srv->pidpath[0]) {
		fprintf(stderr, "pidpath has been set at %s\n", srv->pidpath);
		exit(1);
	}

	if (!path || !path[0] || strlen(path) > SRV_PATH_SIZE - 1) {
		fprintf(stderr, "illegal pid path: %s\n", path);
		exit(1);
	}

	strncpy(srv->pidpath, path, SRV_PATH_SIZE - 1);
}


